package com.uhealin;

import java.io.IOException;
import java.util.ArrayList;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class HotelReducer extends Reducer<Hotel, NullWritable, Text, NullWritable>{

	@Override
	protected void reduce(Hotel key, Iterable<NullWritable> vals,
			Reducer<Hotel, NullWritable, Text, NullWritable>.Context context) throws IOException, InterruptedException {
		// TODO 자동 생성된 메소드 스텁
		int count=0;
		
		ArrayList<Hotel> hotels=new ArrayList<Hotel>();
		
		for(NullWritable val:vals){
			if(count==0&&key.getType().get()!=0)break;
			
			if(key.getType().get()==0){
				try {
					hotels.add((Hotel)key.clone());
				} catch (CloneNotSupportedException e) {
					// TODO 자동 생성된 catch 블록
					//e.printStackTrace();
				}
			}else{
				
				for(Hotel h: hotels){
					double dis=Hotel.calDistance(h, key);
				   if(dis<=Hotel.RADIU){
					   Text text=new Text(h.getId()+"\t"+key.getId()+"\t"+(int)dis);
					   context.write(text,NullWritable.get());
				   }
				}
				
			}
			++count;
		}
		
		
	}
	
	

}
